/**
 * Project Name:rx-java
 * File Name:RxJavaHelloWorld.java
 * Package Name:top.xiaocao.rx.java.hello.world
 * Date:2024年2月11日下午6:05:21
 * Copyright (c) 2024, zhengchong.wan@minicore.com.hk All Rights Reserved.
 *
*/

package top.xiaocao.rx.java.hello.world;
/**
 * @className:RxJavaHelloWorld 
 * @comment: 注释信息 
 * @date:     2024年2月11日 下午6:05:21 
 * @author   zhengchong.wan
 * @since    V1.0.0	 
 */

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class RxJavaHelloWorld {

	public static void main(String[] args) throws InterruptedException {
		var helloWord = new RxJavaHelloWorld();
		var observable = helloWord.passenger;
		var observer = helloWord.driver;
		observable.subscribe(observer);
		
		System.out.println("---------------" + Thread.currentThread().getId());
		
		Observable.create(new ObservableOnSubscribe<String>() {

			@Override
			public void subscribe(@NonNull ObservableEmitter<@NonNull String> e) throws Throwable {
				if (!e.isDisposed()) {
					System.out.println("Observable emit 1," + Thread.currentThread().getId());
		            e.onNext("1");
		            System.out.println("Observable emit 2," + Thread.currentThread().getId());
		            e.onNext("2");
		            System.out.println("Observable emit 3," + Thread.currentThread().getId());
		            e.onNext("3");
		            e.onComplete();
		            System.out.println("Observable emit 4," + Thread.currentThread().getId());
		            e.onNext("4");
				}
			}
		}).subscribeOn(Schedulers.io())
		.observeOn(Schedulers.newThread())
		.subscribe(new Observer<String>() {

			private int i;
	        private Disposable mDisposable;
			
			@Override
			public void onSubscribe(@NonNull Disposable d) {
				mDisposable = d;
				System.out.println("开始采用onSubscribe连接," + Thread.currentThread().getId());
			}

			@Override
			public void onNext(@NonNull String t) {
				i++;
	            if (i == 2) {
	                // 在RxJava 2.x 中，新增的Disposable可以做到切断的操作，让Observer观察者不再接收上游事件
	                mDisposable.dispose();
	            }
	            System.out.println("Observer receive " + t + "," + Thread.currentThread().getId());
			}

			@Override
			public void onError(@NonNull Throwable e) {
				System.out.println("对error事件做出响应" + Thread.currentThread().getId());
			}

			@Override
			public void onComplete() {
				System.out.println("对complete事件做出响应" + Thread.currentThread().getId());
			}
		});
		
		Thread.sleep(10000);
	}
	
	private Observable<String> passenger = Observable.create(new ObservableOnSubscribe<String>() {

		@Override
		public void subscribe(@NonNull ObservableEmitter<@NonNull String> e) throws Throwable {
			System.out.println("Observable emit 1");
            e.onNext("1");
            System.out.println("Observable emit 2");
            e.onNext("2");
            System.out.println("Observable emit 3");
            e.onNext("3");
            e.onComplete();
            System.out.println("Observable emit 4");
            e.onNext("4");
		}
	});
	
	private Observer<String> driver = new Observer<String>() {
		
		private int i;
        private Disposable mDisposable;
		
		@Override
		public void onSubscribe(@NonNull Disposable d) {
			mDisposable = d;
			System.out.println("开始采用onSubscribe连接");
		}

		@Override
		public void onNext(@NonNull String t) {
			i++;
            if (i == 2) {
                // 在RxJava 2.x 中，新增的Disposable可以做到切断的操作，让Observer观察者不再接收上游事件
                mDisposable.dispose();
            }
            System.out.println("Observer receive " + t);
		}

		@Override
		public void onError(@NonNull Throwable e) {
			System.out.println("对error事件做出响应");
		}

		@Override
		public void onComplete() {
			System.out.println("对complete事件做出响应");
		}
	};

}

